package com.xq.jltx.rxjavaretrofitdemopro;

import android.content.Context;
import android.graphics.Bitmap;
import android.graphics.BitmapFactory;

import java.util.ArrayList;
import java.util.List;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Action2;
import rx.schedulers.Schedulers;

/**
 * @author jltxseo
 *         Created by junlintianxia on 2016/08/04.
 */
public class SimpleRxJava {

    /**
     *  RxJava简单调用
     * @param isUseSubscriber
     */
    public void startRunRxJava(boolean isUseSubscriber){
        //创建RxJava的观察者，它决定事件触发的时候将有怎样的行为
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onCompleted() {
                System.out.println("Completed!");
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error! : e=>"+e.toString());
            }

            @Override
            public void onNext(String s) {
                System.out.println("Next:"+s);
            }
        };

        //在 RxJava 的 subscribe 过程中，Observer 也总是会先被转换成一个 Subscriber 再使用
        /*
        onStart(): 这是 Subscriber 增加的方法。它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。需要注意的是，如果对准备工作的线程有要求（例如弹出一个显示进度的对话框，这必须在主线程执行）， onStart() 就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法，具体可以在后面的文中看到。
        unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。在这个方法被调用后，Subscriber 将不再接收事件。一般在这个方法调用前，可以使用 isUnsubscribed() 先判断一下状态。 unsubscribe() 这个方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。
         */
        Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onStart() {
                System.out.println("Start!");
            }

            @Override
            public void onCompleted() {
                System.out.println("Completed!");
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error! : e=>"+e.toString());
            }

            @Override
            public void onNext(String s) {
                System.out.println("Next:"+s);
            }
        };
        //第一种，创建Observable 即被观察者，决定什么时候触发事件以及触发怎样的事件
        Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("Hello");
                subscriber.onNext("Hi");
                subscriber.onNext("RxJava");
                subscriber.onCompleted();
            }
        });
        //创建了 Observable 和 Observer 之后，再用 subscribe() 方法将它们联结起来，整条链子就可以工作了。
        if(isUseSubscriber){
            observable.subscribe(subscriber);
        }else {
            observable.subscribe(observer);
        }
    }
    /**
     * 合并多个Observables的发射物
     * 使用Merge操作符你可以将多个Observables的输出合并，就好像它们是一个单个的Observable一样。
     * Merge可能会让合并的Observables发射的数据交错
     * merge合并多个子任务的时候
     */
    public void merge() {
        Observable<Integer> odds = Observable.just(1, 3, 5);
        odds.subscribeOn(Schedulers.io())
                .doOnSubscribe(new Action0() {     //这里刚开始执行订阅时进的回调，线程控制是由它的下面晚上切换
                    @Override
                    public void call() {
                        System.out.println("odds.doOnSubscribe.Thread:"+Thread.currentThread().getName());
                    }
                })
                .subscribeOn(AndroidSchedulers.mainThread())
                .observeOn(AndroidSchedulers.mainThread())  //这里将订阅者的回调线程切换到主线程
                .doOnNext(new Action1<Integer>() {          //监听单个任务的在执行公共onNext函数前调用，可以通过这里对各个子任务进行监听onNext函数
                    @Override
                    public void call(Integer integer) {
                        System.out.println("odds.doOnNext.Thread:"+Thread.currentThread().getName());
                    }
                })
                .doOnCompleted(new Action0() {
                    @Override
                    public void call() {
                        System.out.println("odds.doOnCompleted.Thread:"+Thread.currentThread().getName());
                    }
                })
                .doOnError(new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        System.out.println("odds.doOnError.Thread:"+Thread.currentThread().getName());

                    }
                });
        Observable<Integer> evens = Observable.just(2, 4, 6);
        evens.subscribeOn(Schedulers.io())
                .doOnSubscribe(new Action0() {
                    @Override
                    public void call() {
                        System.out.println("evens.doOnSubscribe.Thread:"+Thread.currentThread().getName());
                    }
                })
                .subscribeOn(AndroidSchedulers.mainThread())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("evens.doOnNext.Thread:"+Thread.currentThread().getName());
                    }
                })
                .doOnCompleted(new Action0() {
                    @Override
                    public void call() {
                        System.out.println("evens.doOnCompleted.Thread:"+Thread.currentThread().getName());
                    }
                })
                .doOnError(new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        System.out.println("evens.doOnError.Thread:"+Thread.currentThread().getName());
                    }
                });
/*        evens.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                logger("evens complete.");
            }

            @Override
            public void onError(Throwable e) {
                logger("evens:Error: " + e.getMessage());
            }

            @Override
            public void onNext(Integer integer) {
                logger("evens: " + integer);
            }
        });*/
        List<Observable<Integer>> tasks = new ArrayList<Observable<Integer>>();
        tasks.add(odds);
        tasks.add(evens);
        Subscription subscription = Observable.merge(tasks)
/*                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {

                    }
                }, new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {

                    }
                }, new Action0() {
                    @Override
                    public void call() {

                    }
                });*/
                .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onNext(Integer item) {
                        System.out.println("merge.doOnNext.Thread:"+Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(Throwable error) {
                        System.out.println("merge.onError.Thread:"+Thread.currentThread().getName());
                    }

                    @Override
                    public void onCompleted() {
                        System.out.println("merge.onCompleted.Thread:"+Thread.currentThread().getName());
                    }
                });
    }
    /**
     * 支持不完整定义的回调的订阅
     */
    public void startRunRxJava2(boolean isSwitch){

        Action1<String> onNextAction = new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNextAction:"+s);
            }
        };

        Action1<Throwable> onErroAction = new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {

            }
        };

        Action0 onCompletedAction = new Action0() {
            @Override
            public void call() {
                System.out.println("Completed!");
            }
        };

        Action2<String,String> onNextAction2 = new Action2<String, String>() {
            @Override
            public void call(String s, String s2) {
                System.out.println("s1:"+s+" s2:"+s2);
            }
        };
        Observable observable = null;
        //经过这两种方式创建的Obervable
        // 将会依次调用：
        // onNext("Hello");
        // onNext("Hi");
        // onNext("RxJava");
        // onCompleted();
        if(isSwitch){
            //第二种，创建Observable 即被观察者，决定什么时候触发事件以及触发怎样的事件just(T...): 将传入的参数依次发送出来
             observable = Observable.just("Hello","Hi","RxJava");
        }else {
            String[] words = {"Hello","Hi","RxJava"};
            observable = Observable.from(words);
        }


        //自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
        observable.subscribe(onNextAction);
        // 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
        observable.subscribe(onNextAction, onErroAction);
        // 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
        observable.subscribe(onNextAction, onErroAction, onCompletedAction);
    }

    /**
     * RxJava线程控制Scheduler
     */
    public void startRunRxJavaScheduler1 (){
        Observable.just(1,2,3,4)
                .subscribeOn(Schedulers.io())// 指定 subscribe() 发生在 IO 线程
                .observeOn(Schedulers.immediate())  //指定 Subscriber 的回调直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("number:" + integer);
                    }
                }, new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        System.out.println("Error! : e=>"+throwable.toString());
                    }
                });
    }

    /**
     * RxJava线程控制Scheduler
     */
    public void startRunRxJavaScheduler2(){
        Integer[] words = {1,2,3,4};
        Observable.from(words)
//                .subscribeOn(Schedulers.io())// 指定 subscribe() 发生在 IO 线程
                .observeOn(Schedulers.immediate())  //指定 Subscriber 的回调直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("number:" + integer);
                    }
                }, new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        System.out.println("Error! : e=>"+throwable.toString());
                    }
                });

    }

    public void startRunRxJavaScheduler3(){
        Observable observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                subscriber.onNext(1);
                subscriber.onNext(2);
                subscriber.onNext(3);
                subscriber.onNext(4);
            }
        });
        observable
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.immediate())
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("number:" + integer);
                    }
                });
    }

    public void startRunRxJavaScheduler4(final Context context, final int resId, final OnRxJavaCallBack onRxJavaCallBack){
        Observable.create(new Observable.OnSubscribe<Bitmap>() {
            @Override
            public void call(Subscriber<? super Bitmap> subscriber) {
                Bitmap bmp = BitmapFactory.decodeResource(context.getResources(),resId);
                if(bmp != null && !bmp.isRecycled()){
                    subscriber.onNext(bmp);
                    subscriber.onCompleted();
                }else {
                    subscriber.onError(new Exception("bmp is invalid"));
                }
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Bitmap>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {
                        if(onRxJavaCallBack != null){
                            onRxJavaCallBack.onError(e);
                        }
                    }

                    @Override
                    public void onNext(Bitmap bitmap) {
                        if(onRxJavaCallBack != null){
                            onRxJavaCallBack.onCompleted(bitmap);
                        }
                    }
                });

    }

    public void startRunRxJavaScheduler5(final Context context, final int resId, final OnRxJavaCallBack onRxJavaCallBack){
        Observable.create(new Observable.OnSubscribe<Bitmap>() {
            @Override
            public void call(Subscriber<? super Bitmap> subscriber) {
                Bitmap bmp = BitmapFactory.decodeResource(context.getResources(),resId);
                if(bmp != null && !bmp.isRecycled()){
                    subscriber.onNext(bmp);
                    subscriber.onCompleted();
                }else {
                    subscriber.onError(new Exception("bmp is invalid"));
                }
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Bitmap>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {
                        if(onRxJavaCallBack != null){
                            onRxJavaCallBack.onError(e);
                        }
                    }

                    @Override
                    public void onNext(Bitmap bitmap) {
                        if(onRxJavaCallBack != null){
                            onRxJavaCallBack.onCompleted(bitmap);
                        }
                    }
                });

    }

    public interface OnRxJavaCallBack{
        public void onCompleted(Object object);
        public void onError(Throwable e);
    }
}
